#include "unp.h"
#include <pthread.h>

#define MAXNITEMS 150000

#define min(a,b)                             \
        (a<b?a:b)  

int nitems;

struct {
  pthread_mutex_t mutex;
  int buf[MAXNITEMS];
  int nput;
  int nval;
} shared ={
  PTHREAD_MUTEX_INITIALIZER
};

struct {
  pthread_mutex_t mutex;
  pthread_cond_t  cond;
  int    nread;
} nread = {
  PTHREAD_MUTEX_INITIALIZER,PTHREAD_COND_INITIALIZER 
};

void *produce(void *), *consume(void *);

int main(int argc,char *argv[])
{
  int i,nthreads,count[MAXNITEMS]; 
  pthread_t tid_produce[MAXNITEMS],tid_consume;

  if(argc < 3)
     err_quit("Usage: pro <#nitems> <#npthread>");

  nitems = min(atoi(argv[1]),MAXNITEMS);  
  nthreads = min(atoi(argv[2]),MAXNITEMS);  

  if(pthread_setconcurrency((nthreads+1)) !=0){
    err_quit("setconcurrency error");
  }

  for(i=0;i<nthreads;i++){
    count[i] = 0;
    if(pthread_create(&tid_produce[i],NULL,produce,&count[i])!=0)
        err_quit("pthread_create error");
  }

  if(pthread_create(&tid_consume,NULL,consume,NULL)!=0)
        err_quit("pthread_create error");

  for(i=0;i<nthreads;i++){
    pthread_join(tid_produce[i],NULL);
  }

  pthread_join(tid_consume,NULL);
  exit(0);
}

void * produce(void *c)
{
  printf("produce create tid:%ld\n",pthread_self());
  for(;;){
     pthread_mutex_lock(&shared.mutex);
     if(shared.nput >= nitems){
       pthread_mutex_unlock(&shared.mutex);
       return (NULL);
     } 
     shared.buf[shared.nput] = shared.nval;
     shared.nput++;
     shared.nval++;
     *((int*)c) +=1;
     pthread_mutex_unlock(&shared.mutex);
     
     int flag;
     pthread_mutex_lock(&nread.mutex);
     flag = (nread.nread==0);
     printf("pthread_cond_signal done!\n");
     nread.nread++; 
     pthread_mutex_unlock(&nread.mutex);

     if(flag)
       pthread_cond_signal(&nread.cond);
  }
}

void consume_wait(int i)
{
  for(;;){
     pthread_mutex_lock(&shared.mutex);
     if(i<shared.nput){
       pthread_mutex_unlock(&shared.mutex);
       return;
     }
     pthread_mutex_unlock(&shared.mutex);
  }
}

void * consume(void *c)
{
  int i;
  printf("consume create tid:%ld\n",pthread_self());
  for(i=0; i<nitems; i++){
    pthread_mutex_lock(&nread.mutex);
    while(nread.nread==0){
         pthread_cond_wait(&nread.cond,&nread.mutex);
         printf("pthread_cond_wait done!\n");
    }
    nread.nread--;
    pthread_mutex_unlock(&nread.mutex);
    //consume_wait(i);
    if(shared.buf[i] != i){
       printf("buff[%d] = %d\n",i,shared.buf[i]);
    }
  }
  return (NULL);
}
